package cn.doitedu.benchmark;

import cn.doitedu.eagle.beans.LogBean;
import com.alibaba.fastjson.JSON;
import org.apache.commons.lang.RandomStringUtils;
import org.apache.commons.lang3.RandomUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.HashMap;
import java.util.Properties;

/**
 * @author 涛哥
 * @nick_name "deep as the sea"
 * @contact qq:657270652 wx:doit_edu
 * @site www.doitedu.cn
 * @date 2021-03-25
 * @desc 压测数据模拟器
 * {
 * "account":"157","appId":"cn.do",
 * "appVersion":"3.0","carrier":"中国移动",
 * "deviceId":"157","deviceType":"mi10",
 * "eventId":"thumbUp","ip":"152.33.68.90",
 * "latitude":56.29385792,"longitude":120.329857234,
 * "netType":"wifi","osName":"android","osVersion":"9.0",
 * "properties":{"productId":"10","pageId":"10"},
 * "releaseChannel":"小米应用",
 * "resolution":"1024*2048",
 * "sessionId":"sessionid157",
 * "timeStamp":1616142957968
 * }
 *
 */
public class GenLogData {

    public static void main(String[] args) throws InterruptedException {

        Properties props = null;
        props = new Properties();
        props.setProperty("bootstrap.servers", "hdp01:9092,hdp02:9092,hdp03:9092");
        props.put("acks", "all");// acks=0 配置适用于实现非常高的吞吐量 , acks=all 这是最安全的模式
        //发送到同一个partition的消息会被先存储在batch中，该参数指定一个batch可以使用的内存大小，单位是 byte。不一定需要等到batch被填满才能发送
        props.put("batch.size", 1024);// 默认16384=16KB
        //生产者在发送消息前等待linger.ms，从而等待更多的消息加入到batch中。如果batch被填满或者linger.ms达到上限，就把batch中的消息发送出去
        props.put("linger.ms", 1);// 当linger.ms>0时，延时性会增加，但会提高吞吐量，因为会减少消息发送频率"
        // props.put(“client.id”, 1);//用于标识发送消息的客户端，通常用于日志和性能指标以及配额
        props.put("buffer.memory", 33554432);// 32MB
        // Snappy压缩技术是Google开发的，它可以在提供较好的压缩比的同时，减少对CPU的使用率并保证好的性能，所以建议在同时考虑性能和带宽的情况下使用。
        // Gzip压缩技术通常会使用更多的CPU和时间，但会产生更好的压缩比，所以建议在网络带宽更受限制的情况下使用
        //props.put("compression.type", "Gzip"); // 默认不压缩，该参数可以设置成snappy、gzip或lz4对发送给broker的消息进行压缩
        // 默认值为0，当设置为大于零的值，客户端会重新发送任何发送失败的消息。
        // 注意，此重试与客户端收到错误时重新发送消息是没有区别的。
        // 在配置max.in.flight.requests.per.connection不等于1的情况下，允许重试可能会改变消息的顺序
        // 因为如果两个批次的消息被发送到同一个分区，第一批消息发送失败但第二批成功，而第一批消息会被重新发送，则第二批消息会先被写入。
        props.put("retries", 1);
        props.put("max.in.flight.requests.per.connection", 2);// 生产者在收到服务器响应之前可以发送的消息个数
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        String[] events = {"pageView", "addCart", "collect", "thumbUp", "share", "search", "adShow", "adClick", "fetchCoupon", "useCoupon", "register"};

        for(int i=0;i<10;i++){
            new Thread(
                    new Runnable() {
                        @Override
                        public void run() {
                            while (true) {
                                LogBean bean = new LogBean();
                                // 决定人数，3个就可能有1000人
                                String account = StringUtils.leftPad(RandomStringUtils.randomNumeric(2),4,"0");
                                bean.setAccount(account);

                                String eventId = events[RandomUtils.nextInt(0, events.length)];
                                bean.setEventId(eventId);
                                bean.setTimeStamp(System.currentTimeMillis());
                                bean.setAppId("cn.do");
                                bean.setAppVersion("3.0");
                                bean.setCarrier("中国移动");
                                bean.setDeviceId(account);
                                bean.setDeviceType("mi10");
                                bean.setIp("152.33.68.90");
                                bean.setLatitude(56.29385792);
                                bean.setLongitude(120.329857234);
                                bean.setSessionId("sessionid" + account);
                                bean.setResolution("1024*2048");
                                bean.setReleaseChannel("小米应用");
                                bean.setNetType("wifi");
                                bean.setOsName("android");
                                bean.setOsVersion("9.0");

                                HashMap<String, String> properties = new HashMap<>();
                                // 页面数量可能达到10种
                                properties.put("pageId", RandomUtils.nextInt(1, 11) + "");
                                if (eventId.equals("addCart") || eventId.equals("collect") || eventId.equals("share") || eventId.equals("thumbUp")) {
                                    properties.put("productId", RandomUtils.nextInt(1, 11) + "");
                                }
                                if (eventId.equals("fetchCoupon") || eventId.equals("useCoupon")) {
                                    properties.put("couponId", RandomUtils.nextInt(1, 11) + "");
                                }
                                if (eventId.equals("adShow") || eventId.equals("adClick")) {
                                    properties.put("adId", RandomUtils.nextInt(1, 11) + "");
                                }

                                if (eventId.equals("search")) {
                                    properties.put("keyword", RandomStringUtils.randomNumeric(1));
                                }

                                bean.setProperties(properties);

                                producer.send(new ProducerRecord<String, String>("app_log", JSON.toJSONString(bean)));

                                try {
                                    // 20ms相当于每秒2k条，相当于2w人同时在线
                                    // 当前配置情况下，完全hold住没有任何反压延迟
                                    // 内存占用约6G（原因，缓存没有清除机制，内存一直在增长）
                                    Thread.sleep(RandomUtils.nextInt(1000,4000));
                                } catch (InterruptedException e) {
                                    e.printStackTrace();
                                }

                            }
                        }
                    }
            ).start();
        }
    }
}
